package drds.binlog.example;

import drds.binlog.client.Connector;
import drds.binlog.client.Connectors;
import drds.binlog.client.impl.SimpleConnector;
import drds.binlog.protocol.Message;

import java.net.InetSocketAddress;
import java.util.concurrent.ArrayBlockingQueue;

public class SimpleCanalClientPermanceTest
{

    public static void main(String args[])
    {
        String destination = "example";
        String ip = "127.0.0.1";
        int batchSize = 1024;
        int count = 0;
        int sum = 0;
        int perSum = 0;
        long start = System.currentTimeMillis();
        long end = 0;
        final ArrayBlockingQueue<Long> queue = new ArrayBlockingQueue<Long>(100);
        try
        {
            final Connector connector = Connectors.newSingleConnector(new InetSocketAddress(ip, 11111),
                    destination,
                    "",
                    "");

            Thread ackThread = new Thread(new Runnable()
            {

                @Override
                public void run()
                {
                    while (true)
                    {
                        try
                        {
                            long batchId = queue.take();
                            connector.ack(batchId);
                        } catch (InterruptedException e)
                        {
                        }
                    }
                }
            });
            ackThread.start();

            ((SimpleConnector) connector).setLazyParseEntry(true);
            connector.connect();
            connector.subscribe();
            while (true)
            {
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getRawEntryList().size();
                sum += size;
                perSum += size;
                count++;
                queue.add(batchId);
                if (count % 10 == 0)
                {
                    end = System.currentTimeMillis();
                    long tps = (perSum * 1000) / (end - start);
                    System.out.println(" total : " + sum + " , current : " + perSum + " , cost : " + (end - start)
                            + " , tps : " + tps);
                    start = end;
                    perSum = 0;
                }
            }
        } catch (Throwable e)
        {
            e.printStackTrace();
        }
    }

}
